package com.gitee.sparta.core.process;

import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.util.concurrent.DefaultThreadFactory;

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.*;


/**
 * 线程池静态工厂
 *
 * @author zyred
 * @since v 0.0.1
 */
public class SpartaThreadPoolFactory {

    /** 超时执行线程池 **/
    private ExecutorService timeoutThreadPool;
    /** 监听线程 **/
    private final ExecutorService listenersThreadPool;
    /** 广播线程池 **/
    private ExecutorService broadcastThreadPool;
    /** 配置文件 **/
    private final SpartaCoreProperties properties;

    /**
     * update 2022-05-06 coreSize 为 0 的时候, 无法创建线程池
     *
     * @param properties 配置文件信息
     */
    public SpartaThreadPoolFactory(SpartaCoreProperties properties) {
        this.properties = properties;
        int coreSize = this.getCoreSize();

        this.listenersThreadPool = new ThreadPoolExecutor(
                coreSize, (2 * coreSize),
                this.properties.getThreads().getKeepAliveTime().getSeconds(),
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new DefaultThreadFactory(this.properties.getThreads().getThreadName()),
                new ThreadPoolExecutor.AbortPolicy()
        );
        if (Duration.ZERO.equals(this.properties.getConsumeWaitTime())) {
            this.timeoutThreadPool = Executors.newCachedThreadPool(
                    new DefaultThreadFactory(SpartaConstant.SPARTA_TIMEOUT_THREAD_NAME)
            );
        }
        // 打开广播开关，创建广播线程池
        if (this.properties.isEnableBroadcast()) {
            this.broadcastThreadPool = new ThreadPoolExecutor(
                    coreSize, (coreSize * 2),
                    this.properties.getThreads().getKeepAliveTime().getSeconds(),
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(),
                    new DefaultThreadFactory("broadcast-thread")
            );
        }
        // 添加延迟消息异步线程
        // carry 线程
        // 队列分发线程
        // 广播分发线程
        // leader read 线程
        // follower read 线程
        // leader write 线程
        // follower write 线程
        // follower heartbeat 线程
    }

    /**
     * 获取监听线程的核心线程数
     *
     * IO 密集型
     *
     * @return  监听线程核心线程储量
     */
    private int getCoreSize() {
        // 根据配置文件确定线程核心线程数量
        Integer coreSize = this.properties.getThreads().getCoreSize();
        int numCpu = Runtime.getRuntime().availableProcessors();
        return (Objects.nonNull(coreSize) && coreSize != 0) ? coreSize : (2 * numCpu);
    }

    public ExecutorService getListenersThreadPool() {
        return this.listenersThreadPool;
    }

    public ExecutorService getTimeoutThreadPool() {
        return timeoutThreadPool;
    }

    public ExecutorService getBroadcastThreadPool() {
        return broadcastThreadPool;
    }
}
